package day3;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.Map;

public class job2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.enableCheckpointing(5000);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("bigdata1:9092")
                .setTopics("order")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setGroupId("day3job1")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        FlinkJedisPoolConfig bigdata1 = new FlinkJedisPoolConfig.Builder()
                .setHost("bigdata1")
                .setPort(6379)
                .setPassword("123456")
                .build();

        /*WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner()*/
        WatermarkStrategy<String> watermarks = WatermarkStrategy.<String>noWatermarks();

        SingleOutputStreamOperator<String> process = env.fromSource(source, watermarks, "day3job1")
                .filter(s -> s.contains("1006") || s.contains("1004"))
                .map(s -> {
                    String[] strings = s.split(",");
                    // 用户名 最终金额 状态
                    return new String[]{strings[2], strings[4], strings[5]};
                }).keyBy(s -> s[0])
                .window(TumblingEventTimeWindows.of(Time.seconds(60)))
                .process(new ProcessWindowFunction<String[], String[], String, TimeWindow>() {
                    ValueState<Integer> money;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        money = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("money", Integer.TYPE));
                    }

                    @Override
                    public void process(String s, ProcessWindowFunction<String[], String[], String, TimeWindow>.Context context, Iterable<String[]> iterable, Collector<String[]> collector) throws Exception {
                        for (String[] strings : iterable) {
                            Integer value = money.value();
                            if (strings[2].equals("1004")) {
                                value += Integer.getInteger(strings[1]);
                            } else value -= Integer.getInteger(strings[1]);
                            money.update(value);
                        }
                        collector.collect(new String[]{s, String.valueOf(money.value())});
                    }
                }).windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessAllWindowFunction<String[], String, TimeWindow>() {
                    ValueState<String> last;
                    MapState<String, String> last1;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        last = getRuntimeContext().getState(new ValueStateDescriptor<String>("last", String.class));
                        last1 = getRuntimeContext().getMapState(new MapStateDescriptor<>("last1", String.class, String.class));
                    }

                    @Override
                    public void process(ProcessAllWindowFunction<String[], String, TimeWindow>.Context context, Iterable<String[]> iterable, Collector<String> collector) throws Exception {
                        int i = 0;
                        for (String[] strings : iterable) {
                            i++;
                            if (i >= 2) {
                                Iterable<String> values = last1.values();
                                for (String value : values) {
                                    if (Integer.getInteger(strings[1]) > Integer.getInteger(value)) {
                                        last1.remove(last1.get(value));
                                        last1.put(strings[0], strings[1]);
                                    }
                                }
                            } else last1.put(strings[0], strings[1]);
                        }
                        Iterator<Map.Entry<String, String>> iterator = last1.iterator();
                        String s = last.value();
                        for (Iterator<Map.Entry<String, String>> it = iterator; it.hasNext(); ) {
                            String key = it.next().getKey();
                            String value = it.next().getValue();
                            s += key + value;
                            last.update(s);
                        }
                        collector.collect(last.value());
                    }
                });
        process.addSink(new RedisSink<>(bigdata1 , new day3.RedisSink()));
    }
}
